package io.mqttpush.mqttserver.msgs;

import io.mqttpush.mqttserver.config.DeviceManager;
import io.mqttpush.mqttserver.handles.Iservice;
import io.netty.buffer.ByteBuf;
import io.netty.channel.Channel;
import io.netty.channel.ChannelHandlerContext;
import io.netty.handler.codec.mqtt.*;
import io.netty.util.ReferenceCountUtil;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.nio.charset.StandardCharsets;
import java.util.List;

/**
 * @author tianzhenjiu
 * @date 2020/4/27 16:22
 */
public class MessageService implements Iservice<MqttMessage> {

    Logger logger = LoggerFactory.getLogger(getClass());


    @Override
    public boolean accept(ChannelHandlerContext context, MqttMessage inMessage) {

        MqttMessageType messageType = inMessage.fixedHeader().messageType();
        return messageType == MqttMessageType.PUBLISH ||
                messageType == MqttMessageType.PUBACK ||
                messageType == MqttMessageType.PUBREC ||
                messageType == MqttMessageType.PUBREL ||
                messageType == MqttMessageType.PUBCOMP;

    }

    @Override
    public List<Object> service(ChannelHandlerContext context, List<Object> outMessage, MqttMessage inMessage) {

        MqttFixedHeader fixedHeader = inMessage.fixedHeader();
        MqttMessageType messageType = fixedHeader.messageType();


        switch (messageType) {

            // 对于QOS=1和QOS=2收到回复报文之后都是相同的处理
            case PUBLISH:
                outMessage.add(publish(context, (MqttPublishMessage) inMessage));
                break;
            case PUBACK:
                // 客户端发布收到
            case PUBREC:
                outMessage.add(pubrec(context, inMessage, fixedHeader.qosLevel()));
                break;
            // 客户端发布释放
            case PUBREL:
                outMessage.add(pubrel(context, inMessage));
                break;
            case PUBCOMP:
                ReferenceCountUtil.release(inMessage);
                break;
            default:
                break;
        }

        return outMessage;
    }


    /**
     *
     * 构建一个格式化的消息
     * @param context
     * @param messagepub
     */
    private  ByteBuf  createSendByteBuf(ChannelHandlerContext context,MqttPublishMessage messagepub){


        String  from=DeviceManager.getDeviceManager().deviceId(context.channel());
        String  toTopic=messagepub.variableHeader().topicName();
        ByteBuf sendContent=messagepub.content();



       ByteBuf byteBuf=context.alloc().buffer();


       byteBuf.writeInt(from.length());
       byteBuf.writeBytes(from.getBytes(StandardCharsets.UTF_8));

       byteBuf.writeInt(toTopic.length());
       byteBuf.writeBytes(toTopic.getBytes(StandardCharsets.UTF_8));

       byteBuf.writeInt(sendContent.readableBytes());
       byteBuf.writeBytes(sendContent);

       return byteBuf;
    }
    /**
     * 处理客户端的发布请求 根据客户端请求的QOS级别 发送相应的响应 把相应的消息存储在数据库 现在的 客户端标识是 发送方
     *
     * @param messagepub
     */
    private MqttMessage publish(ChannelHandlerContext context, MqttPublishMessage messagepub) {

        Channel channel=context.channel();

        logger.info("recv->{},{}", channel.id(), messagepub.variableHeader().topicName());

        ByteBuf newContent=createSendByteBuf(context,messagepub);
        messagepub=messagepub.replace(newContent);



        MessageSender.getInstance().sendMessage(channel.id().asShortText(), messagepub.variableHeader().topicName(), messagepub);

        MqttFixedHeader msgHeader = messagepub.fixedHeader();
        MqttQoS mqttQoS = msgHeader.qosLevel();

        MqttPublishVariableHeader header = messagepub.variableHeader();

        switch (mqttQoS) {
            case EXACTLY_ONCE:
                MqttMessageIdVariableHeader connectVariableHeader = MqttMessageIdVariableHeader.from(header.packetId());
                return new MqttPubAckMessage(
                        new MqttFixedHeader(MqttMessageType.PUBREC, false, MqttQoS.EXACTLY_ONCE, false, 0)
                        , connectVariableHeader);
            default:
                break;

        }

        return new MqttMessage(
                new MqttFixedHeader(MqttMessageType.PUBACK, false, MqttQoS.AT_LEAST_ONCE, false, 0), MqttMessageIdVariableHeader.from(header.packetId()));


    }

    /**
     * @param context
     * @param messagepub 处理客户端过来的发布释放 在消息池里根据客户端标识和消息id面拿到一个消息 并且把这个消息发送出去 现在的 客户端标识是 发送方的
     */
    private MqttPubAckMessage pubrel(ChannelHandlerContext context, MqttMessage messagepub) {

        MqttMessageIdVariableHeader variableHeader = (MqttMessageIdVariableHeader) messagepub.variableHeader();

        MqttFixedHeader fixedHeader = new MqttFixedHeader(MqttMessageType.PUBCOMP, false, MqttQoS.AT_LEAST_ONCE, false,
                0);

        MqttPubAckMessage ackMessage = new MqttPubAckMessage(fixedHeader,
                MqttMessageIdVariableHeader.from(variableHeader.messageId()));


        return ackMessage;
    }

    /**
     * 处理客户端 发布收到
     * <p>
     * 对客户端发送发布释放 根据 客户端收到的messageid 找到相应的message 并且 存储到消息记录里面 现在的 客户端标识是 接受方的
     * 现在的messageid是数据库里面的主键id 最后移除重发队列 防止消息重发
     *
     * @param context
     * @param messagepub
     */
    private MqttPubAckMessage pubrec(ChannelHandlerContext context, MqttMessage messagepub, MqttQoS mqttQoS) {

        /**
         * 如果是qos=2的还需要发消息释放的报文
         */
        MqttMessageIdVariableHeader variableHeader = null;
        MqttFixedHeader fixedHeader = null;

        switch (mqttQoS) {
            case EXACTLY_ONCE:
                variableHeader = (MqttMessageIdVariableHeader) messagepub.variableHeader();
                fixedHeader = new MqttFixedHeader(MqttMessageType.PUBREL, false, MqttQoS.AT_LEAST_ONCE, false, 0);
                break;
            case AT_LEAST_ONCE:
                variableHeader = (MqttMessageIdVariableHeader) messagepub.variableHeader();
                fixedHeader = new MqttFixedHeader(MqttMessageType.PUBACK, false, MqttQoS.AT_LEAST_ONCE, false, 0);
                break;

            default:
                break;
        }

        return new MqttPubAckMessage(fixedHeader,
                MqttMessageIdVariableHeader.from(variableHeader.messageId()));
    }
}
